package com.dayu.rabbitmq.confirm;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

import com.dayu.rabbitmq.util.MQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;

/**
 * @author dayu
 * @date 2019/02/20
 * @Describe 单条信息confirm牛
 */
public class ConfirmSend3 {
    private static String QUEUE_NAME = "test_confirm3";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = MQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 注意：同一个队列只支持一个模式,如果同一个队列已经选择了事务模式就不能选择confirm模式
        channel.confirmSelect();

        // 未确认消息的标识
        final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        channel.addConfirmListener(new ConfirmListener() {

            /**
             * 重试1s，2s，5s，。。。
             */
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                    System.out.println("==handleNack == multiple==true");
                } else {
                    System.out.println("==handleNack == multiple==false");
                    confirmSet.remove(deliveryTag);
                }
            }

            /**
             * 没问题的handleAck
             */
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                    System.out.println("==handleAck == multiple==true");
                } else {
                    System.out.println("==handleAck == multiple==false");
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        String msg = "hello confirm3 msg";
        while (true) {
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            confirmSet.add(nextPublishSeqNo);
        }

        // String msg = "hello confirm msg";
        // // 批量发送
        // for (int i = 0; i < 10; i++) {
        // msg = msg + "=" + i;
        // channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        // }
        //
        // // 批量确认
        // if (!channel.waitForConfirms()) {
        // System.out.println("msg send failed");
        // } else {
        // System.out.println("msg send success");
        // }
        // channel.close();
        // connection.close();

    }
}
